package guda.mrpc.core.codec;

import guda.mrpc.core.RpcRequest;
import guda.mrpc.core.RpcResponse;
import guda.mrpc.core.serialize.ObjectSerialize;
import guda.mrpc.core.transport.Hearbeat;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;

/**
 * Created by well on 17/6/3.
 */
public class RpcDecoder extends ReplayingDecoder<RpcDecoder.State> {

    private static final int MAX_BODY_SIZE=1024*1024*5;

    private ObjectSerialize objectSerialize;

    public RpcDecoder(ObjectSerialize objectSerialize){
        super(State.HEADER_MAGIC);
        this.objectSerialize=objectSerialize;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        MsgHeader msgHeader = new MsgHeader();
        switch (state()) {
            case HEADER_MAGIC:
                checkMagic(in.readShort());
                checkpoint(State.HEADER_MSG_TYPE);
            case HEADER_MSG_TYPE:
                msgHeader.setMsgType(in.readShort());
                checkpoint(State.HEADER_BODY_LENGTH);
            case HEADER_BODY_LENGTH:
                msgHeader.setBodyLength(in.readInt());
                checkpoint(State.BODY);
            case BODY:
                switch (msgHeader.getMsgType()) {
                    case MsgHeader.HEARTBEAT:
                        out.add(Hearbeat.NullHearbeat.instance);
                        break;
                    case MsgHeader.REQUEST: {
                        int length = checkBodyLength(msgHeader.getBodyLength());
                        byte[] bytes = new byte[length];
                        in.readBytes(bytes);
                        RpcRequest deserialize = objectSerialize.deserialize(bytes, RpcRequest.class);
                        out.add(deserialize);
                        break;
                    }
                    case MsgHeader.RESPONSE: {
                        int length = checkBodyLength(msgHeader.getBodyLength());
                        byte[] bytes = new byte[length];
                        in.readBytes(bytes);
                        RpcResponse deserialize = objectSerialize.deserialize(bytes, RpcResponse.class);
                        out.add(deserialize);
                        break;
                    }
                    default:
                        throw new RuntimeException("decode illegal sign ");
                }
                checkpoint(State.HEADER_MAGIC);
        }
    }

    private static void checkMagic(short magic)  {
        if (magic != MsgHeader.MAGIC) {
            throw new RuntimeException("unkown magic");
        }
    }

    private static int checkBodyLength(int size)  {
        if (size > MAX_BODY_SIZE) {
            throw new RuntimeException("body length exceed 5m");
        }
        return size;
    }

    enum State {
        HEADER_MAGIC,
        HEADER_MSG_TYPE,
        HEADER_BODY_LENGTH,
        BODY
    }
}
